package rpc

import (
	"context"
	"encoding/json"
	"fmt"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"net"
	"platon-tools/go-service/common/util"
	"platon-tools/go-service/event"
	go_logger "platon-tools/go-service/go-logger"
	"platon-tools/go-service/module/gcloud_k8s"
	rpc_pb "platon-tools/go-service/protocol/rpc_pb"
	"sync"
	"time"
)

const (
	timerOut = 30 * time.Second
	version  = 4
)

type Config struct {
	CenterUrl       string                  `json:"center_url"`
	RpcUrl          string                  `json:"rpc_url"`
	K8sConfig       gcloud_k8s.GCloudConfig `json:"k8s_config"`
	RegisterURL     string                  `json:"register_url"`
	PrometheusSD    string                  `json:"prometheus_sd"`
	MonitorNodePort string                  `json:"monitor_node_port"`
}

type MonitorService struct {
	k8s    *gcloud_k8s.GCPlatonK8s
	client rpc_pb.CentreServiceClient

	config *Config
	feed   *event.Event

	addCh chan gcloud_k8s.PodEle

	ctx      context.Context
	doCancel context.CancelFunc

	customSD []util.CustomSD

	mu sync.Mutex
}

func (ms *MonitorService) GetPodsList(ctx context.Context, request *rpc_pb.RpcRequest) (*rpc_pb.PodsListResponse, error) {
	podlist := ms.k8s.GetPods()
	data, err := json.Marshal(podlist)
	if err != nil {
		return nil, err
	}

	var res rpc_pb.PodsListResponse

	res.Pods = data
	res.Version = version

	return &res, nil
}

func (ms *MonitorService) AddNode(ctx context.Context, request *rpc_pb.AddPodRequest) (*rpc_pb.ResultRequest, error) {
	name := request.GetName()
	chain := request.ChainType.String()
	go ms.addNode(name, chain)

	return &rpc_pb.ResultRequest{
		Code:    0,
		Err:     nil,
		Version: version,
	}, nil
}

func (ms *MonitorService) addNode(podName, chain string) {
	if err := ms.k8s.AddNode(podName, ms.config.K8sConfig.ClusterName, chain); err != nil {
		go_logger.Logger.Error("add node failed", zap.Error(err))
		ms.feed.Send(event.NewPod{
			Err:     err,
			PodName: podName,
		})

		return
	}

	pod := ms.k8s.GetPodFromNameByOnline(podName)
	if pod == nil {
	}
	ms.addCh <- *pod

	ms.feed.Send(event.NewPod{
		Err:     nil,
		PodName: podName,
		Item: event.PodItem{
			Name:       pod.Label.Name,
			IP:         pod.SvcIP,
			Status:     pod.Status,
			StartTime:  pod.StartTime,
			CreateTime: pod.CreateTime,
			App:        pod.Name,
			NodeId:     "",
		},
	})
}

func NewMonitorService(ctx context.Context, config Config, feed *event.Event) *MonitorService {
	self, cancel := context.WithCancel(ctx)
	ms := MonitorService{
		k8s:      new(gcloud_k8s.GCPlatonK8s),
		ctx:      self,
		doCancel: cancel,
		config:   &config,
		feed:     feed,
		addCh:    make(chan gcloud_k8s.PodEle, 5),
	}

	return &ms
}

func (ms *MonitorService) Start() error {
	// todo 需要谷歌云支持
	//if err := ms.k8s.Start(ms.ctx, ms.config.K8sConfig); err != nil {
	//	go_logger.Logger.Error("monitor service start failed", zap.Error(err))
	//	return err
	//}
	//
	//sd, err := util.OpenSDConfig(ms.config.PrometheusSD)
	//if err != nil {
	//	go_logger.Logger.Error("want prometheus SD file failed", zap.Error(err))
	//	return err
	//}
	//
	//ms.customSD = sd;

	go ms.eventLoop()
	go ms.rpcLoop()
	go ms.centreRpcLoop()
	// todo 需要谷歌云支持
	//go ms.monitorNodesLoop()

	return nil
}

func (ms *MonitorService) monitorNodesLoop() {
	defer ms.doCancel()

	pods := ms.k8s.GetPods()
	ch := make(chan string, len(pods))

	l := map[string]context.CancelFunc{}
	mu := sync.Mutex{}

	runFunc := func(pod gcloud_k8s.PodEle, ch chan string) {
		c, cancel := context.WithCancel(ms.ctx)
		defer cancel()
		mu.Lock()
		if _, ok := l[pod.Name]; !ok {
			l[pod.Name] = cancel
		}
		mu.Unlock()

		mn := NewMonitorNode(c, &pod, ms.k8s, ms.feed, 60)
		if err := mn.Start(); err != nil {
			go_logger.Logger.Error("monitor node start failed", zap.Error(err), zap.String("node", pod.Name))
		}

	}
	update := false
	for _, pod := range pods {
		//// todo 测试代码
		//if i == 0 {
		//	pod.NodeID = "3c189d0490d1479b68bfdf50d8ff07b5e8c765edfd1fca6cd916b6489f8fc6c246882d23a42a6861f0d86085bda3e40108edb428864ba94d93dadd73d9f3ebc0"
		//} else if i == 1 {
		//	pod.NodeID = "bcb7ff0bbcd56e181b8983704edee13be44b06ceadbf61487321c8a7f4644c30715876c079e00a8679b91312b17bc03241b1696cdd62518d585c57e34407d6c0"
		//}
		//ip := fmt.Sprintf("%s:%s", pod.SvcIP, ms.config.MonitorNodePort)
		if r := ms.updateSd(pod); r {
			update = true
		}
		go runFunc(pod, ch)
	}

	if update {
		if err := util.UpdateYmlConfig(ms.customSD, ms.config.PrometheusSD); err != nil {
			go_logger.Logger.Error("update prometheus sd file failed", zap.Error(err))
		}
	}

	select {
	case <-ms.ctx.Done():
		return
	case pod, ok := <-ms.addCh:
		if ok {
			if r := ms.updateSd(pod); r {
				if err := util.UpdateYmlConfig(ms.customSD, ms.config.PrometheusSD); err != nil {
					go_logger.Logger.Error("update prometheus sd file failed", zap.Error(err))
				}
			}
			go runFunc(pod, ch)
		}
	case name, ok := <-ch:
		if ok {
			go_logger.Logger.Error("pod quit", zap.String("name", name))
		}
	}

}

func (ms *MonitorService) updateSd(pod gcloud_k8s.PodEle) bool {
	ms.mu.Lock()
	defer ms.mu.Unlock()

	ok := false
	for _, sd := range ms.customSD {
		if sd.Labels["hostname"] == pod.NodeID {
			ok = true
		}
	}

	if !ok {
		ip := fmt.Sprintf("%s:%s", pod.SvcIP, ms.config.MonitorNodePort)
		l := map[string]string{"hostname": pod.NodeID}
		sd := util.CustomSD{
			Targets: []string{ip},
			Labels:  l,
		}
		ms.customSD = append(ms.customSD, sd)
	}

	return ok
}

func (ms *MonitorService) rpcLoop() {
	defer ms.doCancel()
	go_logger.Logger.Debug("rpcLoop rcp run loop")
	lis, err := net.Listen("tcp", ms.config.RpcUrl)
	if err != nil {
		return
	}

	grpcServer := grpc.NewServer()
	rpc_pb.RegisterMonitorServiceServer(grpcServer, ms)
	grpcServer.Serve(lis)
}

func (ms *MonitorService) eventLoop() {
	defer ms.doCancel()
	defer func() {
		go_logger.Logger.Sugar().Debug("event loop exit")
	}()
	go_logger.Logger.Debug("eventLoop rcp run loop")
	ch := make(chan event.MonitorEvent, 5)
	defer close(ch)
	sub := ms.feed.Subscribe(ch)
	defer sub.Unsubscribe()

	for {
		select {
		case <-ms.ctx.Done():
			return
		case e, ok := <-ch:
			if ok {
				if ms.client != nil {
					go_logger.Logger.Sugar().Info("MsgType_QueryCandidateList ", "event name: ", e.Name)
					ctx, _ := context.WithTimeout(ms.ctx, timerOut)
					if _, err := ms.client.Push(ctx, &rpc_pb.PushRequest{
						Type:    e.Name,
						Data:    e.Data,
						Version: version,
					}); err != nil {
						go_logger.Logger.Error("push request failed", zap.Error(err))
					}
				}
			}
		case err := <-sub.Err():
			go_logger.Logger.Error("subscribe failed", zap.Error(err))
			return
		}
	}
}

func (ms *MonitorService) centreRpcLoop() {
	go_logger.Logger.Debug("centreRpcLoop run loop")
	ticker := time.NewTicker(timerOut)
	defer ticker.Stop()
	init := false
	var conn *grpc.ClientConn
	var err error
	for {
		select {
		case <-ms.ctx.Done():
			return
		case <-ticker.C:
			if !init {
				ctx, _ := context.WithTimeout(ms.ctx, timerOut)
				conn, err = grpc.DialContext(ctx, ms.config.CenterUrl, grpc.WithInsecure())
				if err != nil {
					go_logger.Logger.Error("Dial centre failed", zap.Error(err), zap.String("monitor url", ms.config.CenterUrl))
				} else {
					ms.client = rpc_pb.NewCentreServiceClient(conn)
					rr := rpc_pb.RegisterRequest{
						Url:     ms.config.RegisterURL,
						Version: version,
					}
					ctx, _ := context.WithTimeout(ms.ctx, timerOut)
					if _, err := ms.client.RegisterMonitor(ctx, &rr); err != nil {
						go_logger.Logger.Error("Register monitor failed", zap.Error(err), zap.String("monitor url", ms.config.CenterUrl))
					} else {
						go_logger.Logger.Sugar().Debugf("register monitor ok")
						init = true
					}
				}
			} else {
				ctx, _ := context.WithTimeout(ms.ctx, timerOut)
				if _, err := ms.client.Ping(ctx, &rpc_pb.RpcRequest{Version: version}); err != nil {
					init = false
					if conn != nil {
						conn.Close()
						ms.client = nil
					}
					go_logger.Logger.Error("ping center failed", zap.Error(err))
				}
			}

			ticker.Reset(timerOut)
		}
	}
}
